package org.apache.flink.runtime.rest.messages.taskmanager;

import java.util.Collection;
import java.util.List;

import javax.annotation.Nullable;

import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.resourcemanager.ResourceProviderInfoWithSlots;
import org.apache.flink.runtime.resourceprovider.ResourceProviderMemoryConfiguration;
import org.apache.flink.runtime.rest.messages.ResourceProfileInfo;
import org.apache.flink.runtime.rest.messages.json.ResourceIDDeserializer;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerMetricsInfo.GarbageCollectorInfo;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonIgnore;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import org.apache.flink.util.Preconditions;

/** Message containing base information about a {@link TaskExecutor} and more detailed metrics. */
public class TaskManagerDetailsInfo extends ResourceProviderInfo {

    public static final String FIELD_NAME_ALLOCATED_SLOTS = "allocatedSlots";

    @JsonProperty(FIELD_NAME_ALLOCATED_SLOTS)
    private final Collection<SlotInfo> allocatedSlots;

    @JsonCreator
    public TaskManagerDetailsInfo(
            @JsonDeserialize(using = ResourceIDDeserializer.class)
                    @JsonProperty(FIELD_NAME_RESOURCE_ID)
                    ResourceID resourceId,
            @JsonProperty(FIELD_NAME_ADDRESS) String address,
            @JsonProperty(FIELD_NAME_JMX_PORT) int jmxPort,
            @JsonProperty(FIELD_NAME_LAST_HEARTBEAT) long lastHeartbeat,
            @JsonProperty(FIELD_NAME_NUMBER_SLOTS) int numberSlots,
            @JsonProperty(FIELD_NAME_NUMBER_AVAILABLE_SLOTS) int numberAvailableSlots,
            @JsonProperty(FIELD_NAME_TOTAL_RESOURCE) ResourceProfileInfo totalResource,
            @JsonProperty(FIELD_NAME_AVAILABLE_RESOURCE) ResourceProfileInfo freeResource,
            @JsonProperty(FIELD_NAME_HARDWARE) HardwareDescription hardwareDescription,
            @JsonProperty(FIELD_NAME_MEMORY) ResourceProviderMemoryConfiguration memoryConfiguration,
            @JsonProperty(FIELD_NAME_BLOCKED) @Nullable Boolean blocked,
            @JsonProperty(FIELD_NAME_ALLOCATED_SLOTS) Collection<SlotInfo> allocatedSlots
            ) {
        super(
                resourceId,
                address,
                jmxPort,
                lastHeartbeat,
                numberSlots,
                numberAvailableSlots,
                totalResource,
                freeResource,
                hardwareDescription,
                memoryConfiguration,
                blocked);

        this.allocatedSlots = Preconditions.checkNotNull(allocatedSlots);

    }

    public TaskManagerDetailsInfo(
            ResourceProviderInfoWithSlots resourceProviderInfoWithSlots) {
        this(
                resourceProviderInfoWithSlots.getTaskManagerInfo().getResourceId(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getAddress(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getJmxPort(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getLastHeartbeat(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getNumberSlots(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getNumberAvailableSlots(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getTotalResource(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getFreeResource(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getHardwareDescription(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getMemoryConfiguration(),
                resourceProviderInfoWithSlots.getTaskManagerInfo().getBlocked(),
                resourceProviderInfoWithSlots.getAllocatedSlots());
    }
    
    @JsonIgnore
    public Collection<SlotInfo> getAllocatedSlots() {
        return allocatedSlots;
    }

}
